package com.walker.util;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Base64;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;

import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

public class AmbariTemplateUtil {

	private String endpoint;
	
	private String clusterName;

	private Map<String, Object> versions;

	private String[] types = new String[] { "HDFS", "HIVE", "MAPREDUCE2", "TEZ", "SPARK", "YARN" };

	private HttpEntity<Map<String, Object>> requestEntity;
	private RestTemplate restTemplate;

	public AmbariTemplateUtil(String key, String endpoint, String clusterName) {

		this.endpoint = endpoint;
		this.clusterName = clusterName;

		try {
			String str = Base64.getEncoder().encodeToString(key.getBytes("utf-8"));

			HttpHeaders headers = new HttpHeaders();
			headers.setContentType(MediaType.APPLICATION_JSON);
			headers.set("Authorization", "Basic " + str);

			Map<String, Object> params = new HashMap<String, Object>();

			requestEntity = new HttpEntity<Map<String, Object>>(params, headers);

			restTemplate = new RestTemplate();

			versions = new HashMap<String, Object>();
			for (String type : types) {
				versions.put(type, "");
			}

			getCurrentVersion();
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
	}

	public List<Map<String, Object>> check() throws UnsupportedEncodingException {

		String[] types = new String[] { "HIVE", "SPARK", "HDFS", "ZOOKEEPER", "TEZ", "YARN", "KAFKA" };

		List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
		
		Set<String> serviceNames=getServiceNames();
		
		for (String type : types) {

			Map<String, Object> map = new HashMap<String, Object>();

			if(serviceNames.contains(type)){
				ResponseEntity<String> response = restTemplate.exchange(endpoint + "/services/" + type, HttpMethod.GET,requestEntity, String.class);
				
				String state = JSONObject.fromObject(response.getBody()).getJSONObject("ServiceInfo").getString("state");

				if ("STARTED".equals(state)) {
					map.put("state", true);
				} else {
					map.put("state", false);
				}

				map.put("info", type + " " + state);
			}else{
				map.put("info", type + " 未安装此服务");
			}
			
			list.add(map);
		}

		return list;
	}

	private Set<String> getServiceNames() {
		ResponseEntity<String> response = restTemplate.exchange(endpoint + "/services/", HttpMethod.GET,requestEntity, String.class);
		
		JSONArray ja=JSONObject.fromObject(response.getBody()).getJSONArray("items");
		
		Set<String> set=new HashSet<String>();
		
		for(int i=0;i<ja.size();i++){
			JSONObject item=ja.getJSONObject(i);
			String serviceName=item.getJSONObject("ServiceInfo").getString("service_name");
			set.add(serviceName);
		}
		
		return set;
	}

	public void getCurrentVersion() {
		ResponseEntity<String> response = restTemplate.exchange(endpoint, HttpMethod.GET, requestEntity, String.class);

		JSONObject desired_service_config_versions = JSONObject.fromObject(response.getBody()).getJSONObject("Clusters")
				.getJSONObject("desired_service_config_versions");

		for (String type : types) {
			if (desired_service_config_versions.keySet().contains(type)) {
				JSONArray HDFS = desired_service_config_versions.getJSONArray(type);
				for (int i = 0; i < HDFS.size(); i++) {
					JSONObject json = HDFS.getJSONObject(i);
					if (clusterName.equals(json.getString("cluster_name"))) {
						versions.put(type, json.getString("service_config_version"));
					}
				}
			}
		}
	}

	public void getHiveInfo(Map<String, Object> infos) {
		ResponseEntity<String> response = restTemplate
				.exchange(endpoint + "/configurations/service_config_versions?service_name=HIVE&service_config_version="
						+ versions.get("HIVE"), HttpMethod.GET, requestEntity, String.class);

		JSONObject items = JSONObject.fromObject(response.getBody()).getJSONArray("items").getJSONObject(0);

		JSONArray configurations = items.getJSONArray("configurations");
		for (int j = 0; j < configurations.size(); j++) {
			JSONObject configuration = configurations.getJSONObject(j);
			String type = configuration.getString("type");
			if ("hive-env".equals(type)) {
				JSONObject properties = configuration.getJSONObject("properties");
				infos.put("hive_database_type", properties.getString("hive_database_type"));
			} else if ("hive-site".equals(type)) {
				JSONObject properties = configuration.getJSONObject("properties");

				infos.put("connction_driver_name", properties.getString("javax.jdo.option.ConnectionDriverName"));
				infos.put("connction_url", properties.getString("javax.jdo.option.ConnectionURL"));
				infos.put("connction_user_name", properties.getString("javax.jdo.option.ConnectionUserName"));
				infos.put("thrift_port", properties.getString("hive.server2.thrift.port"));

				String metastore_uris = properties.getString("hive.metastore.uris");
				infos.put("metastore_host",
						metastore_uris.substring(metastore_uris.indexOf("//") + 2, metastore_uris.lastIndexOf(":")));
			}
		}
	}

	public void getHdfsInfo(Map<String, Object> infos) {
		ResponseEntity<String> response = restTemplate
				.exchange(endpoint + "/configurations/service_config_versions?service_name=HDFS&service_config_version="
						+ versions.get("HDFS"), HttpMethod.GET, requestEntity, String.class);

		JSONObject items = JSONObject.fromObject(response.getBody()).getJSONArray("items").getJSONObject(0);

		JSONArray configurations = items.getJSONArray("configurations");
		for (int j = 0; j < configurations.size(); j++) {
			JSONObject configuration = configurations.getJSONObject(j);
			String type = configuration.getString("type");
			if ("core-site".equals(type)) {
				JSONObject properties = configuration.getJSONObject("properties");

				infos.put("fs_defaultFS", properties.getString("fs.defaultFS"));
			}
		}
	}

	public void getMAPREDUCE2Info(Map<String, Object> infos) {
		ResponseEntity<String> response = restTemplate.exchange(
				endpoint + "/configurations/service_config_versions?service_name=MAPREDUCE2&service_config_version="
						+ versions.get("MAPREDUCE2"),
				HttpMethod.GET, requestEntity, String.class);

		JSONObject items = JSONObject.fromObject(response.getBody()).getJSONArray("items").getJSONObject(0);

		JSONArray configurations = items.getJSONArray("configurations");
		for (int j = 0; j < configurations.size(); j++) {
			JSONObject configuration = configurations.getJSONObject(j);
			String type = configuration.getString("type");
			if ("mapred-site".equals(type)) {
				JSONObject properties = configuration.getJSONObject("properties");

				infos.put("mapreduce_job_queuename", properties.getString("mapreduce.job.queuename"));
			}
		}
	}

	public void getTezInfo(Map<String, Object> infos) {
		ResponseEntity<String> response = restTemplate
				.exchange(endpoint + "/configurations/service_config_versions?service_name=TEZ&service_config_version="
						+ versions.get("TEZ"), HttpMethod.GET, requestEntity, String.class);

		JSONArray items = JSONObject.fromObject(response.getBody()).getJSONArray("items");

		if (items.isEmpty()) {
			infos.put("tez_queue_name", "");
		} else {
			JSONObject item = items.getJSONObject(0);
			JSONArray configurations = item.getJSONArray("configurations");
			for (int j = 0; j < configurations.size(); j++) {
				JSONObject configuration = configurations.getJSONObject(j);
				String type = configuration.getString("type");
				if ("tez-site".equals(type)) {
					JSONObject properties = configuration.getJSONObject("properties");

					infos.put("tez_queue_name", properties.getString("tez.queue.name"));
				}
			}
		}

	}

	public void getSparkInfo(Map<String, Object> infos) {
		ResponseEntity<String> response = restTemplate.exchange(
				endpoint + "/configurations/service_config_versions?service_name=SPARK&service_config_version="
						+ versions.get("SPARK"),
				HttpMethod.GET, requestEntity, String.class);

		JSONArray items = JSONObject.fromObject(response.getBody()).getJSONArray("items");

		if (items.isEmpty()) {
			infos.put("spark_yarn_queue", "");
		} else {
			JSONObject item = items.getJSONObject(0);

			JSONArray configurations = item.getJSONArray("configurations");
			for (int j = 0; j < configurations.size(); j++) {
				JSONObject configuration = configurations.getJSONObject(j);
				String type = configuration.getString("type");
				if ("spark-defaults".equals(type)) {
					JSONObject properties = configuration.getJSONObject("properties");

					infos.put("spark_yarn_queue", properties.getString("spark.yarn.queue"));
				}
			}
		}
	}

	public String getHostsIp(String hostName) {
		ResponseEntity<String> response = restTemplate.exchange(endpoint + "/hosts/" + hostName, HttpMethod.GET,requestEntity, String.class);

		return JSONObject.fromObject(response.getBody()).getJSONObject("Hosts").getString("ip");
	}
}
